package com.second.util;

import com.typesafe.config.Config;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;

public class HbaseUtils {

  public static final String TABLE_NAME_USER = "topic_user";
  public static final String FAMILY_USER = "user";
  public static final String FAMILY_SALE = "sale";
  public static final String FAMILY_BEHAVIOR = "behavior";
  public static final String QUALIFIER_USER_AGE = "age_level";
  public static final String QUALIFIER_USER_OCCUPATION = "occupation";
  public static final String QUALIFIER_USER_CITY = "new_user_class_level";
  public static final String QUALIFIER_SLAE_PVALUE = "pvalue_level";
  public static final String QUALIFIER_SLAE_SHOP = "shopping_level";
  public static final String QUALIFIER_BEHAVIOR_RV_7 = "last_7_day_review";
  public static final String QUALIFIER_BEHAVIOR_RV_14 = "last_14_day_review";
  public static final String QUALIFIER_BEHAVIOR_BUY_7 = "last_7_day_buy";
  public static final String QUALIFIER_BEHAVIOR_BUY_14 = "last_14_day_buy";

  public static final String TABLE_NAME_AD = "topic_ad";
  public static final String FAMILY_AD = "adgroup";
  public static final String QUALIFIER_AD_DIS_1 = "last_1_day_display";
  public static final String QUALIFIER_AD_DIS_7 = "last_7_day_display";
  public static final String QUALIFIER_AD_CLK_1 = "last_1_day_clicked";
  public static final String QUALIFIER_AD_CLK_7 = "last_7_day_clicked";

  public static Configuration getHBaseConfig() {
    Configuration conf = HBaseConfiguration.create();
    Config hbaseConf = GetConfig.getRoot("hbase");
    conf.set("hbase.zookeeper.quorum", hbaseConf.getString("host"));
    conf.set("zookeeper.znode.parent", hbaseConf.getString("znode.parent"));

    return conf;
  }

  public static Result get(String name, String key) throws IOException {
    Connection connection = ConnectionFactory.createConnection(getHBaseConfig());

    Table table = connection.getTable(TableName.valueOf(name));
    Get get = new Get(Bytes.toBytes(key));
    Result result = table.get(get);

    table.close();
    connection.close();

    return result;
  }

}

